package com.bigdata.exam.cg

import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Durations, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

object XiaGuanzi {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
      .setMaster("local[2]")
      .setAppName("xiaguanzi")
    val sc = new SparkContext(conf)
    sc.setLogLevel("error")


    val ssc = new StreamingContext(sc, Durations.seconds(3))
    ssc.checkpoint("data/checkpoint")

    val input: DStream[String] = ssc.socketTextStream("node01", 8888)

    val numbers: DStream[(String, Int)] = input.map(line => {
      val arr: Array[String] = line.split(" ")
      (arr(0), arr(2).toInt)
    })

    val result: DStream[(String, Int)] = numbers.updateStateByKey((newValue: Seq[Int], running: Option[Int]) => {
      Option(newValue.sum + running.getOrElse(0))
    })

    result.print()

    ssc.start()
    ssc.awaitTermination()
    ssc.stop()
  }
}
